package com.yox.rpc.transport.netty.client;

import com.yox.rpc.entity.RpcRequest;
import com.yox.rpc.entity.RpcResponse;
import com.yox.rpc.enumeration.RpcError;
import com.yox.rpc.exception.RpcException;
import com.yox.rpc.factory.SingletonFactory;
import com.yox.rpc.loadBalance.LoadBalancer;
import com.yox.rpc.loadBalance.RandomLoadBalancer;
import com.yox.rpc.registry.NacosServiceDiscover;
import com.yox.rpc.registry.ServiceDiscover;
import com.yox.rpc.serializer.CommonSerializer;
import com.yox.rpc.transport.RpcClient;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;
import java.util.concurrent.CompletableFuture;

/**
 * Netty客户端
 * 使用静态代码块初始化Netty客户端，利用Channel将RpcRequest对象传到服务端并设置监听，发送后会立刻返回，异步等待服务器端返回的结果。
 * 利用 AttributeKey 的get()方法阻塞获得返回结果。具体实现就是以 rpcResponse 作为key，以返回的对象作为value，
 * 作为Map对象绑定在Channel上，使用get()方法就能获得返回结果，具体代码在 NettyClientHandler 中。
 */
public class NettyClient implements RpcClient {

    protected static final Logger logger = LoggerFactory.getLogger(NettyClient.class);

    private final ServiceDiscover serviceDiscover;
    private final CommonSerializer serializer;
    private static final EventLoopGroup group;
    private static final Bootstrap bootstrap;
    private final UnprocessedRequests unprocessedRequests;

    static {
        group = new NioEventLoopGroup();
        bootstrap = new Bootstrap();
        bootstrap.group(group)
                .channel(NioSocketChannel.class);
    }
    public NettyClient() {
        // 以默认序列化器调用构造函数, 随机负载均衡
        this(DEFAULT_SERIALIZER, new RandomLoadBalancer());
    }
    public NettyClient(LoadBalancer loadBalancer) {
        this(DEFAULT_SERIALIZER, loadBalancer);
    }
    public NettyClient(Integer serializerID) {
        this(serializerID, new RandomLoadBalancer());
    }
    public NettyClient(Integer serializerID, LoadBalancer loadBalancer) {
        this.serviceDiscover = new NacosServiceDiscover(loadBalancer);
        this.serializer = CommonSerializer.getByCode(serializerID);
        this.unprocessedRequests = SingletonFactory.getInstance(UnprocessedRequests.class);
    }

    @Override
    public CompletableFuture<RpcResponse> sendRequest(RpcRequest rpcRequest) {
        if (serializer == null) {
            logger.error("未设置序列化器");
            throw new RpcException(RpcError.SERVICE_NOT_FOUND);
        }

        CompletableFuture<RpcResponse> resultFuture = new CompletableFuture<>();
        try {
            // 从Nacos注册中心中获取服务地址ip
            InetSocketAddress ip = serviceDiscover.lookupService(rpcRequest.getInterfaceName());
            // 创建Netty连接通道
            Channel channel = ChannelProvider.getChannel(ip, serializer);
            logger.info("客户端连接到服务器 {}:{}", ip.getAddress(), ip.getPort());

            if (channel.isActive()) {
                // 将新请求放入未处理完的请求map中
                unprocessedRequests.put(rpcRequest.getRequestID(), resultFuture);
                // 向服务端发送请求，并设置监听
                channel.writeAndFlush(rpcRequest).addListener((ChannelFutureListener)future1 -> {
                    if (future1.isSuccess()) {
                        logger.info(String.format("客户端发送信息：%s", rpcRequest.toString()));
                    } else {
                        future1.channel().close();
                        resultFuture.completeExceptionally(future1.cause());
                        logger.error("发送信息到时候出错:", future1.cause());
                    }
                });
            } else {
                group.shutdownGracefully();
                return null;
            }
        } catch (Exception e){
            // 将请求从map中移除
            unprocessedRequests.remove(rpcRequest.getRequestID());
            logger.error("发送消息的时候出错：", e);
            // interrupt()这里作用是给受阻塞的当前线程发出一个中断信号，让当前线程退出阻塞状态，好继续执行然后结束
            Thread.currentThread().interrupt();
        }

        return resultFuture;
    }

}
